package priv.pfz.paxos.common;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 基于事件的延迟队列，用于网络异步通信
 * @author pengfangzhou
 * @date 2022/2/6 19:00
 */
@Slf4j
public class EventQueue implements Runnable {
    private final DelayQueue<Event> queue = new DelayQueue<>();
    private final ExecutorService executorService = Executors.newSingleThreadExecutor();

    public EventQueue() {
        executorService.submit(this);
    }

    @Override
    public void run() {
        while (true) {
            try {
                Event event = queue.take();
                event.getRunnable().run();
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    /**
     * 添加事件
     * @param runnable 事件触发执行的操作
     * @param delay  delay毫秒之后触发
     */
    public void add(Runnable runnable, long delay) {
        long effectTime = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(delay);
        queue.add(new Event(runnable, effectTime));
    }
}
